package com.atguigu.hadoop.examples.mapreduce;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapred.ClusterStatus;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.mapreduce.lib.partition.InputSampler;
import org.apache.hadoop.mapreduce.lib.partition.TotalOrderPartitioner;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.net.URI;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

public class Sort<K, V> extends Configured
        implements Tool
{
    public static final String REDUCES_PER_HOST = "mapreduce.sort.reducesperhost";
    private Job job = null;

    static int printUsage() {
        System.out.println("sort [-r <reduces>] [-inFormat <input format class>] [-outFormat <output format class>] [-outKey <output key class>] [-outValue <output value class>] [-totalOrder <pcnt> <num samples> <max splits>] <input> <output>");

        ToolRunner.printGenericCommandUsage(System.out);
        return 2;
    }

    public int run(String[] args)
            throws Exception
    {
        Configuration conf = getConf();
        JobClient client = new JobClient(conf);
        ClusterStatus cluster = client.getClusterStatus();
        int num_reduces = (int)(cluster.getMaxReduceTasks() * 0.9D);
        String sort_reduces = conf.get("mapreduce.sort.reducesperhost");
        if (sort_reduces != null)
        {
            num_reduces = cluster.getTaskTrackers() *
                    Integer.parseInt(sort_reduces);
        }

        Class inputFormatClass = SequenceFileInputFormat.class;

        Class outputFormatClass = SequenceFileOutputFormat.class;

        Class outputKeyClass = BytesWritable.class;
        Class outputValueClass = BytesWritable.class;
        List otherArgs = new ArrayList();
        InputSampler.Sampler sampler = null;
        for (int i = 0; i < args.length; i++) {
            try {
                if ("-r".equals(args[i])) {
                    num_reduces = Integer.parseInt(args[(++i)]);
                } else if ("-inFormat".equals(args[i]))
                {
                    inputFormatClass = Class.forName(args[(++i)])
                            .asSubclass(InputFormat.class);
                } else if ("-outFormat".equals(args[i]))
                {
                    outputFormatClass = Class.forName(args[(++i)])
                            .asSubclass(OutputFormat.class);
                } else if ("-outKey".equals(args[i]))
                {
                    outputKeyClass = Class.forName(args[(++i)])
                            .asSubclass(WritableComparable.class);
                } else if ("-outValue".equals(args[i]))
                {
                    outputValueClass = Class.forName(args[(++i)])
                            .asSubclass(Writable.class);
                } else if ("-totalOrder".equals(args[i])) {
                    double pcnt = Double.parseDouble(args[(++i)]);
                    int numSamples = Integer.parseInt(args[(++i)]);
                    int maxSplits = Integer.parseInt(args[(++i)]);
                    if (0 >= maxSplits) maxSplits = 2147483647;
                    sampler = new InputSampler.RandomSampler(pcnt, numSamples, maxSplits);
                }
                else {
                    otherArgs.add(args[i]);
                }
            } catch (NumberFormatException except) {
                System.out.println("ERROR: Integer expected instead of " + args[i]);
                return printUsage();
            } catch (ArrayIndexOutOfBoundsException except) {
                System.out.println("ERROR: Required parameter missing from " + args[(i - 1)]);

                return printUsage();
            }
        }

        this.job = Job.getInstance(conf);
        this.job.setJobName("sorter");
        this.job.setJarByClass(Sort.class);

        this.job.setMapperClass(Mapper.class);
        this.job.setReducerClass(Reducer.class);

        this.job.setNumReduceTasks(num_reduces);

        this.job.setInputFormatClass(inputFormatClass);
        this.job.setOutputFormatClass(outputFormatClass);

        this.job.setOutputKeyClass(outputKeyClass);
        this.job.setOutputValueClass(outputValueClass);

        if (otherArgs.size() != 2) {
            System.out.println("ERROR: Wrong number of parameters: " + otherArgs
                    .size() + " instead of 2.");
            return printUsage();
        }
        FileInputFormat.setInputPaths(this.job, (String)otherArgs.get(0));
        FileOutputFormat.setOutputPath(this.job, new Path((String)otherArgs.get(1)));

        if (sampler != null) {
            System.out.println("Sampling input to effect total-order sort...");
            this.job.setPartitionerClass(TotalOrderPartitioner.class);
            Path inputDir = FileInputFormat.getInputPaths(this.job)[0];
            FileSystem fs = inputDir.getFileSystem(conf);
            inputDir = inputDir.makeQualified(fs.getUri(), fs.getWorkingDirectory());
            Path partitionFile = new Path(inputDir, "_sortPartitioning");
            TotalOrderPartitioner.setPartitionFile(conf, partitionFile);
            InputSampler.writePartitionFile(this.job, sampler);
            URI partitionUri = new URI(partitionFile.toString() + "#_sortPartitioning");

            this.job.addCacheFile(partitionUri);
        }

        System.out.println("Running on " + cluster
                .getTaskTrackers() + " nodes to sort from " +
                FileInputFormat.getInputPaths(this.job)[
                        0] + " into " +
                FileOutputFormat.getOutputPath(this.job) +
                " with " + num_reduces + " reduces.");

        Date startTime = new Date();
        System.out.println("Job started: " + startTime);
        int ret = this.job.waitForCompletion(true) ? 0 : 1;
        Date end_time = new Date();
        System.out.println("Job ended: " + end_time);
        System.out.println("The job took " +
                (end_time
                        .getTime() - startTime.getTime()) / 1000L + " seconds.");
        return ret;
    }

    public static void main(String[] args)
            throws Exception
    {
        int res = ToolRunner.run(new Configuration(), new Sort(), args);
        System.exit(res);
    }

    public Job getResult()
    {
        return this.job;
    }
}